/*
 *  Copyright (c) 2005-2009, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 *  WSO2 Inc. licenses this file to you under the Apache License,
 *  Version 2.0 (the "License"); you may not use this file except
 *  in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 *
 */
package org.wso2.carbon.registry.core.jdbc.dao;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.wso2.carbon.registry.core.CollectionImpl;
import org.wso2.carbon.registry.core.RegistryConstants;
import org.wso2.carbon.registry.core.ResourceIDImpl;
import org.wso2.carbon.registry.core.ResourceImpl;
import org.wso2.carbon.registry.core.dao.RatingsDAO;
import org.wso2.carbon.registry.core.dao.ResourceDAO;
import org.wso2.carbon.registry.core.dataaccess.DAOManager;
import org.wso2.carbon.registry.core.exceptions.RegistryException;
import org.wso2.carbon.registry.core.jdbc.DatabaseConstants;
import org.wso2.carbon.registry.core.jdbc.dataaccess.JDBCDatabaseTransaction;
import org.wso2.carbon.registry.core.jdbc.dataobjects.RatingDO;
import org.wso2.carbon.registry.core.session.CurrentSession;
import org.wso2.carbon.registry.core.utils.RegistryUtils;
import org.wso2.carbon.utils.DBUtils;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

/**
 * An implementation of the {@link RatingsDAO} to store ratings on a JDBC-based database.
 */
public class JDBCRatingsDAO implements RatingsDAO {

    private static final Log log = LogFactory.getLog(JDBCRatingsDAO.class);
    private ResourceDAO resourceDAO;
    protected static final Object ADD_RATING_LOCK = new Object();

    /**
     * Default constructor
     *
     * @param daoManager instance of the data access object manager.
     */
    public JDBCRatingsDAO(DAOManager daoManager) {
        this.resourceDAO = daoManager.getResourceDAO();
    }

    public void addRating(ResourceImpl resourceImpl, String userID, int rating)
            throws RegistryException {

        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();

        PreparedStatement ps1 = null, ps2 = null, ps3 = null;
        ResultSet result = null;
        try {

            String sql1 =
                    "INSERT INTO REG_RATING (REG_RATING, REG_USER_ID, REG_RATED_TIME, " +
                            "REG_TENANT_ID) VALUES (?,?,?,?)";
            String sql2 = "SELECT MAX(REG_ID) FROM REG_RATING";
            String dbProductName = conn.getMetaData().getDatabaseProductName();
            boolean returnsGeneratedKeys = DBUtils.canReturnGeneratedKeys(dbProductName);
            if (returnsGeneratedKeys) {
                ps1 = conn.prepareStatement(sql1, new String[]{DBUtils
                        .getConvertedAutoGeneratedColumnName(dbProductName,
                                DatabaseConstants.ID_FIELD)});
            } else {
                ps1 = conn.prepareStatement(sql1);
            }
            ps1.setInt(1, rating);
            ps1.setString(2, userID);
            long now = System.currentTimeMillis();
            ps1.setDate(3, new Date(now));
            ps1.setInt(4, CurrentSession.getTenantId());
            if (returnsGeneratedKeys) {
                ps1.executeUpdate();
                result = ps1.getGeneratedKeys();
            } else {
                synchronized (ADD_RATING_LOCK) {
                    ps1.executeUpdate();
                    ps2 = conn.prepareStatement(sql2);
                    result = ps2.executeQuery();
                }
            }
            if (result.next()) {
                int rateID = result.getInt(1);

                String sql3 = "INSERT INTO REG_RESOURCE_RATING (REG_RATING_ID, REG_PATH_ID, " +
                        "REG_RESOURCE_NAME, REG_TENANT_ID) " +
                        "VALUES(?,?,?,?)";
                ps3 = conn.prepareStatement(sql3);

                ps3.setInt(1, rateID);
                ps3.setInt(2, resourceImpl.getPathID());
                ps3.setString(3, resourceImpl.getName());
                ps3.setInt(4, CurrentSession.getTenantId());

                ps3.executeUpdate();

            }

        } catch (SQLException e) {

            String msg = "Failed to rate resource " + resourceImpl.getPath() +
                    " with rating " + rating + ". " + e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                try {
                    if (result != null) {
                        result.close();
                    }
                } finally {
                    try {
                        if (ps1 != null) {
                            ps1.close();
                        }
                    } finally {
                        try {
                            if (ps2 != null) {
                                ps2.close();
                            }
                        } finally {
                            if (ps3 != null) {
                                ps3.close();
                            }
                        }
                    }
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }
    }

    public void copyRatings(ResourceImpl fromResource, ResourceImpl toResource)
            throws RegistryException {
        if (fromResource == null || toResource == null || fromResource.getPath() == null
                || fromResource.getPath().equals(toResource.getPath())) {
            // no special copying needed
            return;
        }
        RatingDO[] ratingDOs = getResourceRatingDO(fromResource);
        addRatings(toResource, ratingDOs);
    }

    public void addRatings(ResourceImpl resource, RatingDO[] ratingDOs)
            throws RegistryException {
        for (RatingDO ratingDO : ratingDOs) {
            addRating(resource, ratingDO.getRatedUserName(), ratingDO.getRating());
        }
    }

    public void updateRating(ResourceImpl resourceImpl, int rateID, int rating)
            throws RegistryException {

        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();

        PreparedStatement ps = null;
        try {

            String sql =
                    "UPDATE REG_RATING SET REG_RATING=?, REG_RATED_TIME=? WHERE REG_ID=? " +
                            "AND REG_TENANT_ID=?";

            long now = System.currentTimeMillis();

            ps = conn.prepareStatement(sql);
            ps.setInt(1, rating);
            ps.setDate(2, new Date(now));
            ps.setInt(3, rateID);
            ps.setInt(4, CurrentSession.getTenantId());

            ps.executeUpdate();

        } catch (SQLException e) {

            String msg = "Failed to update the rating id " + rateID + " of resource " +
                    resourceImpl.getPath() +
                    " to value " + rating + ". " + e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }
    }
    public void removeRating(ResourceImpl resourceImpl , int ratingID) throws RegistryException {

        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();

        PreparedStatement ps1 = null, ps2 = null;
        try {
            String sql1 =
                    "DELETE FROM REG_RESOURCE_RATING WHERE REG_RATING_ID = ? AND REG_TENANT_ID=?";
            ps1 = conn.prepareStatement(sql1);

            String sql2 = "DELETE FROM REG_RATING WHERE REG_ID = ? AND REG_TENANT_ID=?";
            ps2 = conn.prepareStatement(sql2);

            ps1.setLong(1, ratingID);
            ps1.setInt(2, CurrentSession.getTenantId());

            ps2.setLong(1, ratingID);
            ps2.setInt(2, CurrentSession.getTenantId());
            ps1.addBatch();
            ps2.addBatch();

            try {
                ps1.executeBatch();
                ps2.executeBatch();
            } catch (SQLException e) {
                ps1.clearBatch();
                ps2.clearBatch();
                // the exception will be handled in the next catch block
                throw e;
            }

        } catch (SQLException e) {

            String msg = "Failed to remove rating on resource " + resourceImpl.getPath() + ". " +
                    e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                try {
                    if (ps1 != null) {
                        ps1.close();
                    }
                } finally {
                    if (ps2 != null) {
                        ps2.close();
                    }
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }
    }

    public int getRateID(ResourceImpl resourceImpl, String userID) throws RegistryException {

        RatingDO ratingDO = getRatingDO(resourceImpl, userID);
        return ratingDO.getRatingID();
    }

    public float getAverageRating(ResourceImpl resourceImpl) throws RegistryException {

        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();

        int sumRating;
        int countRating;

        PreparedStatement ps = null;
        ResultSet result1 = null;
        try {
            if (resourceImpl instanceof CollectionImpl) {
                String sql = "SELECT SUM(R.REG_RATING) FROM REG_RATING R, REG_RESOURCE_RATING RR " +
                        "WHERE RR.REG_PATH_ID=? AND RR.REG_RESOURCE_NAME IS NULL AND " +
                        "RR.REG_RATING_ID=R.REG_ID AND R.REG_TENANT_ID=? AND RR.REG_TENANT_ID=?";

                ps = conn.prepareStatement(sql);
                ps.setInt(1, resourceImpl.getPathID());
                ps.setInt(2, CurrentSession.getTenantId());
                ps.setInt(3, CurrentSession.getTenantId());
            } else {
                String sql = "SELECT SUM(R.REG_RATING) FROM REG_RATING R, REG_RESOURCE_RATING RR " +
                        "WHERE RR.REG_PATH_ID=? AND RR.REG_RESOURCE_NAME = ? AND " +
                        "RR.REG_RATING_ID=R.REG_ID AND R.REG_TENANT_ID=? AND RR.REG_TENANT_ID=?";

                ps = conn.prepareStatement(sql);
                ps.setInt(1, resourceImpl.getPathID());
                ps.setString(2, resourceImpl.getName());
                ps.setInt(3, CurrentSession.getTenantId());
                ps.setInt(4, CurrentSession.getTenantId());
            }

            result1 = ps.executeQuery();

            sumRating = 0;
            if (result1.next()) {
                sumRating = result1.getInt(1);
            }

        } catch (SQLException e) {

            String msg = "Failed to get sum of all ratings on resource " +
                    resourceImpl.getPath() + ". " + e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                try {
                    if (result1 != null) {
                        result1.close();
                    }
                } finally {
                    if (ps != null) {
                        ps.close();
                    }
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }

        ps = null;
        ResultSet result2 = null;
        try {

            if (resourceImpl instanceof CollectionImpl) {
                String sql =
                        "SELECT COUNT(R.REG_RATING) FROM REG_RATING R, REG_RESOURCE_RATING RR " +
                                "WHERE RR.REG_PATH_ID=? AND RR.REG_RESOURCE_NAME IS NULL AND " +
                                "RR.REG_RATING_ID=R.REG_ID AND R.REG_TENANT_ID=? " +
                                "AND RR.REG_TENANT_ID=?";

                ps = conn.prepareStatement(sql);
                ps.setInt(1, resourceImpl.getPathID());
                ps.setInt(2, CurrentSession.getTenantId());
                ps.setInt(3, CurrentSession.getTenantId());
            } else {
                String sql =
                        "SELECT COUNT(R.REG_RATING) FROM REG_RATING R, REG_RESOURCE_RATING RR " +
                                "WHERE RR.REG_PATH_ID=? AND RR.REG_RESOURCE_NAME = ? AND " +
                                "RR.REG_RATING_ID=R.REG_ID AND R.REG_TENANT_ID=? " +
                                "AND RR.REG_TENANT_ID=?";

                ps = conn.prepareStatement(sql);
                ps.setInt(1, resourceImpl.getPathID());
                ps.setString(2, resourceImpl.getName());
                ps.setInt(3, CurrentSession.getTenantId());
                ps.setInt(4, CurrentSession.getTenantId());
            }

            result2 = ps.executeQuery();

            countRating = 0;
            if (result2.next()) {
                countRating = result2.getInt(1);
            }

        } catch (SQLException e) {

            String msg = "Failed to get ratings count on resource " +
                    resourceImpl.getPath() + ". " + e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                try {
                    if (result2 != null) {
                        result2.close();
                    }
                } finally {
                    if (ps != null) {
                        ps.close();
                    }
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }

        float averageRating = 0;
        if (countRating > 0) {
            averageRating = (float) sumRating / countRating;
        }

        return averageRating;
    }

    public RatingDO getRatingDO(ResourceImpl resourceImpl, String userID) throws RegistryException {

        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();

        PreparedStatement ps = null;
        ResultSet result = null;
        RatingDO ratingDO = new RatingDO();
        try {
            if (resourceImpl instanceof CollectionImpl) {
                String sql = "SELECT R.REG_ID, R.REG_RATING, R.REG_RATED_TIME " +
                        "FROM REG_RATING R, REG_RESOURCE_RATING RR " +
                        "WHERE RR.REG_PATH_ID=? AND RR.REG_RESOURCE_NAME IS NULL AND " +
                        "RR.REG_RATING_ID=R.REG_ID AND R.REG_USER_ID=? AND " +
                        "R.REG_TENANT_ID=? AND RR.REG_TENANT_ID=?";

                ps = conn.prepareStatement(sql);
                ps.setInt(1, resourceImpl.getPathID());
                ps.setString(2, userID);
                ps.setInt(3, CurrentSession.getTenantId());
                ps.setInt(4, CurrentSession.getTenantId());
            } else {
                String sql = "SELECT R.REG_ID, R.REG_RATING, R.REG_RATED_TIME " +
                        "FROM REG_RATING R, REG_RESOURCE_RATING RR " +
                        "WHERE RR.REG_PATH_ID=? AND RR.REG_RESOURCE_NAME = ? AND " +
                        "RR.REG_RATING_ID=R.REG_ID AND R.REG_USER_ID=? AND " +
                        "R.REG_TENANT_ID=? AND RR.REG_TENANT_ID=?";

                ps = conn.prepareStatement(sql);
                ps.setInt(1, resourceImpl.getPathID());
                ps.setString(2, resourceImpl.getName());
                ps.setString(3, userID);
                ps.setInt(4, CurrentSession.getTenantId());
                ps.setInt(5, CurrentSession.getTenantId());
            }

            result = ps.executeQuery();
            if (result.next()) {
                ratingDO.setRating(result.getInt(DatabaseConstants.RATING_FIELD));
                ratingDO.setRatedTime(new java.util.Date(
                        result.getTimestamp(DatabaseConstants.RATED_TIME_FIELD).getTime()));
                ratingDO.setRatedUserName(userID);
                ratingDO.setRatingID(result.getInt(DatabaseConstants.ID_FIELD));
            }
        } catch (SQLException e) {

            String msg = "Failed to get rating on resource " + resourceImpl.getPath() +
                    " done by user " + userID + ". " + e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                try {
                    if (result != null) {
                        result.close();
                    }
                } finally {
                    if (ps != null) {
                        ps.close();
                    }
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }
        return ratingDO;
    }

    public int getRating(ResourceImpl resourceImpl, String userID) throws RegistryException {

        RatingDO ratingDO = getRatingDO(resourceImpl, userID);
        return ratingDO.getRating();
    }

    /**
     * Method to get a rating corresponding to the given id.
     *
     * @param ratingID the id of the rating.
     *
     * @return the rating data object.
     * @throws RegistryException if an error occurs while getting the resource path.
     * @throws SQLException      if an error occurs while getting the rating.
     */
    public RatingDO getRating(long ratingID) throws SQLException, RegistryException {

        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();

        String sql = "SELECT RR.REG_PATH_ID, RR.REG_RESOURCE_NAME, " +
                "R.REG_USER_ID, R.REG_RATING, R.REG_RATED_TIME " +
                "FROM REG_RATING R, REG_RESOURCE_RATING RR WHERE R.REG_ID =? AND " +
                "R.REG_ID = RR.REG_RATING_ID AND R.REG_TENANT_ID=? AND RR.REG_TENANT_ID=?";

        PreparedStatement s = conn.prepareStatement(sql);
        try {
            s.setLong(1, ratingID);
            s.setInt(2, CurrentSession.getTenantId());
            s.setInt(3, CurrentSession.getTenantId());

            ResultSet results = s.executeQuery();
            try {
                if (results.next()) {
                    java.util.Date ratedTime = new java.util.Date(
                            results.getTimestamp(DatabaseConstants.RATED_TIME_FIELD).getTime());

                    RatingDO ratingDAO = new RatingDO();
                    ratingDAO.setRatedUserName(results.getString(DatabaseConstants.USER_ID_FIELD));
                    ratingDAO.setRatedTime(ratedTime);
                    ratingDAO.setRating(results.getInt(DatabaseConstants.RATING_FIELD));
                    // TODO:
                    //ratingDAO.setResourceID(results.getString(DatabaseConstants.AID_FIELD));

                    String resourcePath;

                    int pathId = results.getInt(DatabaseConstants.PATH_ID_FIELD);
                    String resourceName = results.getString(DatabaseConstants.RESOURCE_NAME_FIELD);
                    resourcePath = resourceDAO.getPath(pathId, resourceName, true);
                    if (resourcePath != null) {
                        ratingDAO.setResourcePath(resourcePath);
                    }
                    return ratingDAO;
                }
            } finally {
                if (results != null) {
                    results.close();
                }
            }
        } finally {
            if (s != null) {
                s.close();
            }
        }
        return null;
    }

    public void removeRatings(ResourceImpl resourceImpl) throws RegistryException {

        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();

        PreparedStatement ps1 = null, ps2 = null;
        RatingDO[] ratingDOs = getResourceRatingDO(resourceImpl);
        try {
            String sql1 =
                    "DELETE FROM REG_RESOURCE_RATING WHERE REG_RATING_ID = ? AND REG_TENANT_ID=?";
            ps1 = conn.prepareStatement(sql1);

            String sql2 = "DELETE FROM REG_RATING WHERE REG_ID = ? AND REG_TENANT_ID=?";
            ps2 = conn.prepareStatement(sql2);

            for (RatingDO ratingDO : ratingDOs) {
                ps1.setLong(1, ratingDO.getRatingID());
                ps1.setInt(2, CurrentSession.getTenantId());

                ps2.setLong(1, ratingDO.getRatingID());
                ps2.setInt(2, CurrentSession.getTenantId());
                ps1.addBatch();
                ps2.addBatch();
            }

            if (ratingDOs.length > 0) {
                try {
                    ps1.executeBatch();
                    ps2.executeBatch();
                } catch (SQLException e) {
                    ps1.clearBatch();
                    ps2.clearBatch();
                    // the exception will be handled in the next catch block
                    throw e;
                }
            }

        } catch (SQLException e) {

            String msg = "Failed to remove ratings on resource " + resourceImpl.getPath() + ". " +
                    e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                try {
                    if (ps1 != null) {
                        ps1.close();
                    }
                } finally {
                    if (ps2 != null) {
                        ps2.close();
                    }
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }
    }

    public RatingDO[] getResourceRatingDO(ResourceImpl resourceImpl) throws RegistryException {

        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();

        PreparedStatement ps = null;
        ResultSet result = null;
        ArrayList<RatingDO> ratingDOs = new ArrayList<RatingDO>();
        try {
            if (resourceImpl instanceof CollectionImpl) {
                String sql = "SELECT R.REG_ID, R.REG_RATING, R.REG_USER_ID, R.REG_RATED_TIME " +
                        "FROM REG_RATING R, REG_RESOURCE_RATING RR " +
                        "WHERE RR.REG_PATH_ID=? AND RR.REG_RESOURCE_NAME IS NULL AND " +
                        "RR.REG_RATING_ID=R.REG_ID AND " +
                        "R.REG_TENANT_ID=? AND RR.REG_TENANT_ID=?";

                ps = conn.prepareStatement(sql);
                ps.setInt(1, resourceImpl.getPathID());
                ps.setInt(2, CurrentSession.getTenantId());
                ps.setInt(3, CurrentSession.getTenantId());
            } else {
                String sql = "SELECT R.REG_ID, R.REG_RATING, R.REG_USER_ID, R.REG_RATED_TIME " +
                        "FROM REG_RATING R, REG_RESOURCE_RATING RR " +
                        "WHERE RR.REG_PATH_ID=? AND RR.REG_RESOURCE_NAME = ? AND " +
                        "RR.REG_RATING_ID=R.REG_ID AND R.REG_TENANT_ID=? AND RR.REG_TENANT_ID=?";

                ps = conn.prepareStatement(sql);
                ps.setInt(1, resourceImpl.getPathID());
                ps.setString(2, resourceImpl.getName());
                ps.setInt(3, CurrentSession.getTenantId());
                ps.setInt(4, CurrentSession.getTenantId());
            }

            result = ps.executeQuery();
            while (result.next()) {
                RatingDO ratingDO = new RatingDO();
                ratingDO.setRating(result.getInt(DatabaseConstants.RATING_FIELD));
                ratingDO.setRatedTime(new java.util.Date(
                        result.getTimestamp(DatabaseConstants.RATED_TIME_FIELD).getTime()));
                ratingDO.setRatedUserName(result.getString(DatabaseConstants.USER_ID_FIELD));
                ratingDO.setRatingID(result.getInt(DatabaseConstants.ID_FIELD));

                ratingDOs.add(ratingDO);
            }
        } catch (SQLException e) {

            String msg = "Failed to get rating on resource " + resourceImpl.getPath() +
                    ". " + e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                try {
                    if (result != null) {
                        result.close();
                    }
                } finally {
                    if (ps != null) {
                        ps.close();
                    }
                }
            }
            catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }
        return ratingDOs.toArray(new RatingDO[ratingDOs.size()]);
    }

    public String[] getRatedUserNames(ResourceImpl resourceImpl) throws RegistryException {

        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();

        List<String> userNameList = new ArrayList<String>();

        PreparedStatement ps = null;
        ResultSet results = null;
        try {
            if (resourceImpl instanceof CollectionImpl) {
                String sql = "SELECT R.REG_USER_ID FROM REG_RATING R, REG_RESOURCE_RATING RR " +
                        "WHERE RR.REG_PATH_ID=? AND RR.REG_RESOURCE_NAME IS NULL AND " +
                        "RR.REG_RATING_ID=R.REG_ID AND R.REG_TENANT_ID=? AND RR.REG_TENANT_ID=?";

                ps = conn.prepareStatement(sql);
                ps.setInt(1, resourceImpl.getPathID());
                ps.setInt(2, CurrentSession.getTenantId());
                ps.setInt(3, CurrentSession.getTenantId());
            } else {
                String sql = "SELECT R.REG_USER_ID FROM REG_RATING R, REG_RESOURCE_RATING RR " +
                        "WHERE RR.REG_PATH_ID=? AND RR.REG_RESOURCE_NAME = ? AND " +
                        "RR.REG_RATING_ID=R.REG_ID AND R.REG_TENANT_ID=? AND RR.REG_TENANT_ID=?";

                ps = conn.prepareStatement(sql);
                ps.setInt(1, resourceImpl.getPathID());
                ps.setString(2, resourceImpl.getName());
                ps.setInt(3, CurrentSession.getTenantId());
                ps.setInt(4, CurrentSession.getTenantId());
            }

            results = ps.executeQuery();
            while (results.next()) {
                String ratedUserName = results.getString(DatabaseConstants.USER_ID_FIELD);
                userNameList.add(ratedUserName);
            }

        } catch (SQLException e) {

            String msg = "Failed to users who have rated the resource " +
                    resourceImpl.getPath() + ". " + e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                try {
                    if (results != null) {
                        results.close();
                    }
                } finally {
                    if (ps != null) {
                        ps.close();
                    }
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }

        String[] userNames = new String[userNameList.size()];
        for (int i = 0; i < userNames.length; i++) {
            userNames[i] = userNameList.get(i);
        }

        return userNames;
    }

    public ResourceImpl getResourceWithMinimumData(String path) throws RegistryException {
        return RegistryUtils.getResourceWithMinimumData(path, resourceDAO, false);
    }

    public void moveRatings(ResourceIDImpl source, ResourceIDImpl target) throws RegistryException {
        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();
        PreparedStatement ps = null;
        try {
            if (source.isCollection()) {
                String sql = "UPDATE REG_RESOURCE_RATING SET REG_PATH_ID=? WHERE " +
                        "REG_PATH_ID=? AND REG_RESOURCE_NAME IS NULL AND REG_TENANT_ID=?";
                ps = conn.prepareStatement(sql);
                ps.setInt(1, target.getPathID());
                ps.setInt(2, source.getPathID());
                ps.setInt(3, CurrentSession.getTenantId());
                ps.executeUpdate();
            } else {
                String sql =
                        "UPDATE REG_RESOURCE_RATING SET REG_PATH_ID=?, REG_RESOURCE_NAME=? WHERE " +
                                "REG_PATH_ID=? AND REG_RESOURCE_NAME=? AND REG_TENANT_ID=?";
                ps = conn.prepareStatement(sql);
                ps.setInt(1, target.getPathID());
                ps.setString(2, target.getName());
                ps.setInt(3, source.getPathID());
                ps.setString(4, source.getName());
                ps.setInt(5, CurrentSession.getTenantId());
                ps.executeUpdate();
            }
        } catch (SQLException e) {
            String msg = "Failed to move ratings from  " + source.getPath() +
                    " to " + target.getPath() + ". " + e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }
    }

    public void moveRatingPaths(ResourceIDImpl source, ResourceIDImpl target)
            throws RegistryException {
        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();
        PreparedStatement ps = null;
        try {
            String sql =
                    "UPDATE REG_RESOURCE_RATING SET REG_PATH_ID=? WHERE REG_PATH_ID=? " +
                            "AND REG_TENANT_ID=?";
            ps = conn.prepareStatement(sql);
            ps.setInt(1, target.getPathID());
            ps.setInt(2, source.getPathID());
            ps.setInt(3, CurrentSession.getTenantId());
            ps.executeUpdate();
        } catch (SQLException e) {
            String msg = "Failed to move rating paths from  " + source.getPath() +
                    " to " + target.getPath() + ". " + e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                if (ps != null) {
                    ps.close();
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }
    }
    
    public void removeVersionRatings(long regVersion)
    		throws RegistryException {
        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();

        PreparedStatement ps1 = null, ps2 = null;

        List<Long> taggingIds = getRatingIds(regVersion);
        if (taggingIds == null) {
            return;
        }
        try {

            String sql = "DELETE FROM REG_RESOURCE_RATING WHERE REG_RATING_ID= ? AND REG_TENANT_ID=?";
            ps1 = conn.prepareStatement(sql);

            sql = "DELETE FROM REG_RATING WHERE REG_ID= ? AND REG_TENANT_ID=?";
            ps2 = conn.prepareStatement(sql);

            for (long l : taggingIds) {
                ps1.setLong(1, l);
                ps1.setInt(2, CurrentSession.getTenantId());
                ps2.setLong(1, l);
                ps2.setInt(2, CurrentSession.getTenantId());
                ps1.addBatch();
                ps2.addBatch();
            }

            if (taggingIds.size() > 0) {
                try {
                    ps1.executeBatch();
                    ps2.executeBatch();
                } catch (SQLException e) {
                    ps1.clearBatch();
                    ps2.clearBatch();
                    // the exception will be handled in the next catch block
                    throw e;
                }
            }

        } catch (SQLException e) {

            String msg =
                    "Failed to remove tagging with the version: " + regVersion + ". " + e.getMessage();
            log.error(msg, e);
            throw new RegistryException(msg, e);
        } finally {
            try {
                try {
                    if (ps1 != null) {
                        ps1.close();
                    }
                } finally {
                    if (ps2 != null) {
                        ps2.close();
                    }
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }   	
    }
    
    private List<Long> getRatingIds(long regVersionId) throws RegistryException {

        JDBCDatabaseTransaction.ManagedRegistryConnection conn =
                JDBCDatabaseTransaction.getConnection();
        PreparedStatement ps = null;
        ResultSet results = null;
        List<Long> idList = new ArrayList<Long>();
    	
        try {        	 
        	
        		
        		String sql = "SELECT R.REG_ID FROM REG_RATING R, REG_RESOURCE_RATING RR WHERE " +
        			"RR.REG_VERSION = ? AND RR.REG_TENANT_ID = ? AND RR.REG_TENANT_ID=R.REG_TENANT_ID " +
        			"AND RR.REG_RATING_ID=R.REG_ID";
        	
        		ps = conn.prepareStatement(sql);        		
        		ps.setLong(1, regVersionId);        		
        		ps.setInt(2, CurrentSession.getTenantId());
        	        	
        		results = ps.executeQuery();
        	
        	while(results.next()){
        		idList.add(results.getLong(1));       		
        	}        	
        } catch (Exception ex) {
            String msg = "Failed to retreive the Ratings with the REG_VERSION: " +
                    regVersionId + ". " + ex.getMessage();                
            log.error(msg, ex);
            throw new RegistryException(msg, ex);
        } finally {
            try {
                try {
                    if (results != null) {
                        results.close();
                    }
                } finally {
                    if (ps != null) {
                        ps.close();
                    }
                }
            } catch (SQLException ex) {
                String msg = RegistryConstants.RESULT_SET_PREPARED_STATEMENT_CLOSE_ERROR;
                log.error(msg, ex);
            }
        }        
    	return idList;
    }
}
